package util;

import com.atguigu.function.EventSourceFunction;
import com.atguigu.pojo.Event;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
import org.apache.flink.connector.datagen.source.GeneratorFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class SourceUtil {

    //使用DataGenSource来模拟生成数据
    public static DataGeneratorSource<Event> getSource(){
        return new DataGeneratorSource<Event>(
                new GeneratorFunction<Long, Event>() {

                    String [] users = {"Zhangsan","Lisi","Wangwu","Tom","Jerry"};

                    String [] urls = {"/home","/list","/detail","/pay","/cart"};
                    @Override
                    public Event map(Long value) throws Exception {

                        String user = users[RandomUtils.nextInt(0,users.length)];
                        String url = urls[RandomUtils.nextInt(0,urls.length)];
                        Long ts = System.currentTimeMillis();
                        return new Event (user , url ,ts);
                    }
                },
                Long.MAX_VALUE,
                RateLimiterStrategy.perSecond(1),
                Types.POJO(Event.class)
        );
    }
    //使用自定义SourceFunction来模拟数据
    public static SourceFunction<Event> getSourceFunction(){

        return new EventSourceFunction();
    }

}
